package org.smartboot.sosa.rpc.netty.server;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.smartboot.sosa.rpc.RpcConfig;
import org.smartboot.sosa.rpc.RpcConstant;
import org.smartboot.sosa.rpc.RpcRequest;
import org.smartboot.sosa.rpc.channel.ChannelService;
import org.smartboot.sosa.rpc.netty.NettySession;
import org.smartboot.sosa.rpc.netty.RpcChannelHandler;
import org.smartboot.sosa.rpc.netty.RpcDecoder;
import org.smartboot.sosa.rpc.netty.RpcEncoder;
import org.smartboot.sosa.rpc.process.DisruptorProcessor;
import org.smartboot.sosa.rpc.process.MessageProcessor;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.AttributeKey;

/**
 * 基于Netty实现的RPC服务端
 * 
 * @author Seer
 * @version RpcNettyServer.java, v 0.1 2015年11月24日 下午2:07:36 Seer Exp.
 */
public class RpcNettyServer implements ChannelService {
	private static final Logger LOGGER = LogManager.getLogger(RpcNettyServer.class);
	private EventLoopGroup bossGroup = null; // (1)
	private EventLoopGroup workerGroup = null;
	private ChannelFuture channelFuture;
	/** 消息处理器 */
	private DisruptorProcessor<RpcRequest> processor = null;

	@Override
	public <T> void start(final RpcConfig config, MessageProcessor<T> processor) throws Exception {
		final MessageProcessor<RpcRequest> proce = (MessageProcessor<RpcRequest>) processor;
		// 启动服务器
		bossGroup = new NioEventLoopGroup(); // (1)
		workerGroup = new NioEventLoopGroup();
		ServerBootstrap b = new ServerBootstrap(); // (2)
		b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // (3)
				.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
					@Override
					public void initChannel(SocketChannel ch) throws Exception {
						// 设置编解码处理器
						ch.pipeline().addLast(new RpcDecoder(), new RpcChannelHandler<RpcRequest>(proce))
								.addLast(new RpcEncoder());
						// 注册会话
						NettySession session = new NettySession(ch, config);
						ch.attr(AttributeKey.valueOf(RpcConstant.SESSION_KEY)).set(session);
						if (LOGGER.isDebugEnabled()) {
							LOGGER.debug("init channel " + session);
						}
					}
				}).option(ChannelOption.SO_BACKLOG, 128) // (5)
				.childOption(ChannelOption.SO_KEEPALIVE, true); // (6)

		// Bind and start to accept incoming connections.
		channelFuture = b.bind(config.getPort()).sync(); // (7)

	}

	public void start(RpcConfig config) throws Exception {
		// 启动处理器
		processor = new DisruptorProcessor<RpcRequest>();
		processor.init(config);
		start(config, processor);
	}

	public void shutdown() {
		try {
			if (channelFuture != null) {
				channelFuture.channel().close();
			}
			if (channelFuture != null) {
				channelFuture.channel().closeFuture().sync();
			}
		} catch (Exception e) {
			LOGGER.catching(e);
		} finally {
			if (workerGroup != null) {
				workerGroup.shutdownGracefully();
			}
			if (bossGroup != null) {
				bossGroup.shutdownGracefully();
			}
		}
		try {
			if (processor != null) {
				processor.shutdown();
			}
		} catch (Exception e) {
			LOGGER.catching(e);
		}
	}

}